package com.gmall.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.bean.TableProcessDwd;
import com.gmall.realtime.common.constant.Constant;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.Random;

/**
 * Flink Sink工具类，提供创建常用数据接收器的方法
 * 支持Kafka和Doris两种数据目的地
 */
public class FlinkSinkUtil {

    /**
     * 创建Kafka字符串数据Sink（指定主题）
     * @param topic Kafka主题名称
     * @return Kafka Sink实例
     */
    public static Sink<String> getKafkaSink(String topic) {
        return KafkaSink.<String>builder()
                // 设置Kafka集群地址
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                // 配置记录序列化方式
                .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                        .setTopic(topic)  // 指定目标主题
                        .setValueSerializationSchema(new SimpleStringSchema())  // 使用简单字符串序列化
                        .build())
                // 确保精确一次语义
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 设置事务ID前缀（添加随机数保证唯一性）
                .setTransactionalIdPrefix("hadoop-" + topic + new Random().nextLong())
                // 设置事务超时时间（15分钟）
                .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
                .build();
    }

    /**
     * 创建动态主题的Kafka Sink（根据配置表决定目标主题）
     * 处理Tuple2<数据JSON, 表配置>类型的数据
     * @return Kafka Sink实例
     */
    public static Sink<Tuple2<JSONObject, TableProcessDwd>> getKafkaSink() {
        return KafkaSink.<Tuple2<JSONObject, TableProcessDwd>>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                // 自定义序列化器实现动态主题选择
                .setRecordSerializer(new KafkaRecordSerializationSchema<Tuple2<JSONObject, TableProcessDwd>>() {
                    @Nullable
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(
                            Tuple2<JSONObject, TableProcessDwd> dataWithConfig,
                            KafkaSinkContext context,
                            Long timestamp) {
                        // 从配置信息获取目标表名作为主题
                        String topic = dataWithConfig.f1.getSinkTable();
                        JSONObject data = dataWithConfig.f0;
                        // 移除操作类型字段
                        data.remove("op_type");
                        // 将JSON数据转为字节数组
                        return new ProducerRecord<>(topic, data.toJSONString().getBytes(StandardCharsets.UTF_8));
                    }
                })
                // 确保精确一次语义
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 设置事务ID前缀
                .setTransactionalIdPrefix("hadoop-" + new Random().nextLong())
                // 设置事务超时时间
                .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
                .build();
    }

    /**
     * 创建Doris数据库Sink
     * @param table        目标表标识（database.table格式）
     * @param labelPrefix  Stream Load导入标签前缀
     * @return Doris Sink实例
     */
    public static DorisSink<String> getDorisSink(String table, String labelPrefix) {
        // 配置Stream Load参数
        Properties props = new Properties();
        props.setProperty("format", "json");          // 数据格式为JSON
        props.setProperty("read_json_by_line", "true"); // 每行一条JSON记录

        return DorisSink.<String>builder()
                // Doris读取配置（使用默认配置）
                .setDorisReadOptions(DorisReadOptions.builder().build())
                // Doris连接配置
                .setDorisOptions(DorisOptions.builder()
                        .setFenodes(Constant.DORIS_FE_NODES)  // FE节点地址
                        .setTableIdentifier(table)            // 目标表
                        .setUsername("root")                  // 数据库用户名
                        .setPassword("aaaaaa")                // 数据库密码
                        .build())
                // Doris执行配置
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        .setLabelPrefix(labelPrefix)   // 导入任务标签前缀
                        .disable2PC()                  // 禁用两阶段提交（简化测试）
                        .setBufferCount(3)             // 内存中最大批次个数
                        .setBufferSize(1024 * 1024)    // 批次大小限制（1MB）
                        .setCheckInterval(3000)        // 批次输出间隔（3秒）
                        .setMaxRetries(3)              // 最大重试次数
                        .setStreamLoadProp(props)      // 设置Stream Load参数
                        .build())
                // 使用简单字符串序列化器
                .setSerializer(new SimpleStringSerializer())
                .build();
    }
}
